package com.zfei.firstflink.utils;

import com.zfei.firstflink.model.XexHomeOrder;
import org.apache.flink.api.common.io.DefaultInputSplitAssigner;
import org.apache.flink.api.common.io.RichInputFormat;
import org.apache.flink.api.common.io.statistics.BaseStatistics;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.io.GenericInputSplit;
import org.apache.flink.core.io.InputSplit;
import org.apache.flink.core.io.InputSplitAssigner;
import org.apache.flink.types.Row;

import java.io.IOException;
import java.math.BigDecimal;
import java.sql.*;

/***
 *$com.zfei.firstflink.utils.DataSetMysqlSource
 *@author:jiwenfei
 *
 *@version:
 *
 *createTime:2018/8/22:下午3:20
 *
 *description:
 */
public class DataSetMysqlSource extends RichInputFormat<XexHomeOrder, InputSplit> {
    public DataSetMysqlSource(){

    }
    Connection connection=null;
    PreparedStatement ps=null;
    ResultSet rs=null;
    @Override
    public void configure(Configuration configuration) {

    }

    @Override
    public BaseStatistics getStatistics(BaseStatistics baseStatistics) throws IOException {
        return baseStatistics;
    }

    @Override
    public InputSplit[] createInputSplits(int i) throws IOException {
        return new GenericInputSplit[]{new GenericInputSplit(0, 1)};
    }

    @Override
    public InputSplitAssigner getInputSplitAssigner(InputSplit[] inputSplits) {
        return new DefaultInputSplitAssigner(inputSplits);
    }

    @Override
    public void open(InputSplit inputSplit) throws IOException {
        String driver="com.mysql.jdbc.Driver";
        String url="jdbc:mysql://47.100.10.9:3306/civil_admin";
//        String url="jdbc:mysql://192.168.1.151:3306/civil_administration";
        String userName="zkxbxmysql";
        String passward="Zkxbx-2011!";
//        String passward="zkxbx2011"
        try {
            Class.forName(driver);
            connection= DriverManager.getConnection(url,userName,passward);


            String sql="select order_no,service_object_id,service_provider_id,state,order_type,number,price,spotprice,order_time,business_time,start_time,end_time from civil_admin.xex_home_order order by order_time asc";
            ps=connection.prepareStatement(sql);
            rs= ps.executeQuery();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    @Override
    public boolean reachedEnd() throws IOException {
        try {
            return !rs.next();
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return false;
    }

    @Override
    public XexHomeOrder nextRecord(XexHomeOrder ee) throws IOException {
        try {
            XexHomeOrder order=new XexHomeOrder();
            order.setOrderNo(rs.getString(1));
            order.setServiceObjectId(rs.getString(2));
            order.setServiceProviderId(rs.getString(3));
            order.setState(rs.getInt(4));
            order.setOrderType(rs.getInt(5));
            order.setNumber(rs.getString(6));
            String price=rs.getString("price");
            if (price==null||"".equals(price)||"NaN".equals(price)){
                order.setPrice(new BigDecimal("0"));
            }else {
                try {
                    order.setPrice(new BigDecimal(price));
                    order.setSpotprice(BigDecimal.valueOf(rs.getDouble(8)));
                }catch (Exception e){
                   // e.printStackTrace();
                }
            }

            String tmpTime=null;
            try {
                 tmpTime=rs.getString("order_time");
                if ("0000-00-00 00:00:00".equals(tmpTime)){
                    order.setOrderTime(new Timestamp(System.currentTimeMillis()));
                }else {
                    order.setOrderTime(rs.getTimestamp("order_time"));
                }
            }catch (Exception ex){

                //ex.printStackTrace();
            }

            try {
                tmpTime=rs.getString("start_time");
                if ("0000-00-00 00:00:00".equals(tmpTime)){
                    order.setStartTime(new Timestamp(System.currentTimeMillis()));
                }else {
                    order.setStartTime(rs.getTimestamp("start_time"));
                }
            }catch (Exception ex){

                //ex.printStackTrace();
            }

            try {
                tmpTime=rs.getString("end_time");
                if ("0000-00-00 00:00:00".equals(tmpTime)){
                    order.setEndTime(new Timestamp(System.currentTimeMillis()));
                }else {
                    order.setEndTime(rs.getTimestamp("end_time"));
                }
            }catch (Exception ex){

               // ex.printStackTrace();
            }

            try {
                tmpTime=rs.getString("business_time");
                if ("0000-00-00 00:00:00".equals(tmpTime)){
                    order.setBusinessTime(new Timestamp(System.currentTimeMillis()));
                }else {
                    order.setBusinessTime(rs.getTimestamp("business_time"));
                }
            }catch (Exception ex){

               // ex.printStackTrace();
            }

            return order;
        }catch (Exception e){
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public void close() throws IOException {
       try {
           connection.close();
           ps.close();
           rs.close();
       }catch (Exception e){

       }

    }
}
